package com.dayu.tcp.service;

import com.dayu.tcp.common.util.CollectionUtil;
import io.netty.channel.ChannelHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;

import java.util.ArrayList;
import java.util.EventListener;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author yehangtech
 * @date 2023/11/26 12:52
 * @description 抽象的socket服务，调用者根据自己需求把实现为：服务器模式或者客户端模式
 */
public abstract class Service {
    /**
     * 默认socket类型
     */
    @Getter
    @Setter
    protected SocketType socketType = SocketType.NORMAL;

    /**
     * 默认绑定端口
     */
    @Getter
    @Setter
    protected int port = 10609;

    /**
     * 多ip情况下绑定指定ip
     */
    @Getter
    @Setter
    protected String ip;

    /**
     * 是否启用keepAlive
     */
    @Getter
    @Setter
    protected boolean keepAlive = true;

    /**
     * 是否启用tcpNoDelay
     */
    @Setter
    @Getter
    protected boolean tcpNoDelay = true;

    /**
     * 工作线程池大小
     */
    @Setter
    @Getter
    protected int workerCount;

    /**
     * 是否开启业务处理线程池,默认关闭
     * 如果是服务端，最好启动业务线程池。包括：messageExecutor，channelExecutor，exceptionExecutor
     */
    @Setter
    @Getter
    protected boolean openExecutor = false;

    /**
     * 消息事件业务处理线程池
     */
    @Getter
    protected ExecutorService messageExecutor;

    /**
     * 通道事件业务处理线程池
     */
    protected ExecutorService channelExecutor;

    /**
     * 异常事件业务处理线程池
     */
    protected ExecutorService exceptionExecutor;

    /**
     * 消息事件业务处理线程池最小保持的线程数
     */
    @Setter
    @Getter
    protected int corePoolSize = 10;

    /**
     * 消息事件业务处理线程池最大线程数
     */
    @Setter
    @Getter
    protected int maximumPoolSize = 150;

    /**
     * 消息事件业务处理线程池队列最大值，默认1000000
     */
    @Setter
    @Getter
    protected int queueCapacity = 1000000;

    /**
     * 自定义的业务处理器，会加载到netty的pipeline中
     * key->处理器的名称(自定义)，value->处理器。自定义的业务处理器统一worker线程去执行。
     */
    @Setter
    @Getter
    protected LinkedHashMap<String, ChannelHandler> channelHandlerMap = new LinkedHashMap<>();

    /**
     * 监听事件，注册的时候选择：MessageEventListener，ChannelEventListener，ExceptionEventListener
     * 事件监听的处理会放到对应的线程池中。messageExecutor，channelExecutor，exceptionExecutor
     * 上述的线程池没有开启，则会使用netty的worker线程。服务器模式建议注册上述线程，即openExecutor=true
     */
    @Setter
    @Getter
    protected List<EventListener> eventListenerList = new ArrayList<>();

    /**
     * 服务端与客户端建立链接后 worker线程会调用的回调接口
     */
    @Setter
    @Getter
    protected List<GenericFutureListener<? extends Future<? super Void>>> futureEventListenerList = new ArrayList<>();

    @Setter
    @Getter
    protected EventDispatcher eventDispatcher;

    public Service() {
        //默认工作线程数，CPU密集型任务
        this.workerCount = Runtime.getRuntime().availableProcessors() + 1;
        //添加JVM关闭时的钩子
        Runtime.getRuntime().addShutdownHook(new ShutdownHook(this));
    }

    protected void init() {
        if (openExecutor) {
            this.messageExecutor = new ThreadPoolExecutor(
                    this.corePoolSize,
                    this.maximumPoolSize,
                    60L,
                    TimeUnit.SECONDS,
                    new LinkedBlockingDeque<>(this.queueCapacity),
                    new BasicThreadFactory.Builder().namingPattern("MessageEventProcessor-%d").daemon(true).build(),
                    new ThreadPoolExecutor.AbortPolicy()
            );

            this.channelExecutor = Executors.newCachedThreadPool(
                    new BasicThreadFactory.Builder().namingPattern("ChannelEventProcessor-%d").daemon(true).build()
            );

            this.exceptionExecutor = Executors.newCachedThreadPool(
                    new BasicThreadFactory.Builder().namingPattern("ExceptionEventProcessor-%d").daemon(true).build()
            );
        }
    }

    /**
     * shutdown,关闭钩子由子类实现
     */
    protected abstract void shutdown();

    /**
     * 重新设置-消息事件业务处理线程池
     *
     * @param messageExecutor 消息事件业务处理线程池
     */
    public void setMessageExecutor(@NonNull ExecutorService messageExecutor) {
        ExecutorService preMessageExecutor = this.messageExecutor;
        this.messageExecutor = messageExecutor;
        List<Runnable> taskList = preMessageExecutor.shutdownNow();

        if (CollectionUtil.isNotEmpty(taskList)) {
            taskList.forEach(this.messageExecutor::execute);
        }
    }

    /**
     * @return 线程池中正在执行任务的线程数量
     */
    public int getExecutorActiveCount() {
        if (this.messageExecutor instanceof ThreadPoolExecutor) {
            return ((ThreadPoolExecutor) this.messageExecutor).getActiveCount();
        }
        return -1;
    }

    /**
     * @return 线程池已完成的任务数量，该值小于等于taskCount
     */
    public long getExecutorCompletedTaskCount() {
        if (this.messageExecutor instanceof ThreadPoolExecutor) {
            return ((ThreadPoolExecutor) this.messageExecutor).getCompletedTaskCount();
        }
        return -1L;
    }

    /**
     * @return 线程池曾经创建过的最大线程数量。通过这个数据可以知道线程池是否满过，也就是达到了maximumPoolSize
     */
    public int getExecutorLargestPoolSize() {
        if (this.messageExecutor instanceof ThreadPoolExecutor) {
            return ((ThreadPoolExecutor) this.messageExecutor).getLargestPoolSize();
        }
        return -1;
    }

    /**
     * @return 线程池当前的线程数量
     */
    public int getExecutorPoolSize() {
        if (this.messageExecutor instanceof ThreadPoolExecutor) {
            return ((ThreadPoolExecutor) this.messageExecutor).getPoolSize();
        }
        return -1;
    }

    /**
     * @return 线程池已经执行的和未执行的任务总数
     */
    public long getExecutorTaskCount() {
        if (this.messageExecutor instanceof ThreadPoolExecutor) {
            return ((ThreadPoolExecutor) this.messageExecutor).getTaskCount();
        }
        return -1L;
    }

    /**
     * @return 线程池阻塞队列的size
     */
    public int getExecutorQueueSize() {
        if (this.messageExecutor instanceof ThreadPoolExecutor) {
            return ((ThreadPoolExecutor) this.messageExecutor).getQueue().size();
        }
        return -1;
    }

    /**
     * 新增事件处理
     *
     * @param listener
     */
    public void addEventListener(EventListener listener) {
        this.eventListenerList.add(listener);
    }

    /**
     * 新增回调函数处理
     *
     * @param listener
     */
    public void addFutureEventListener(GenericFutureListener<? extends Future<? super Void>> listener) {
        this.futureEventListenerList.add(listener);
    }

    /**
     * 新增业务处理器-线程安全，可被多个通道共享
     *
     * @param key
     * @param channelHandler
     */
    public void addSharableChannelHandler(String key, ChannelHandler channelHandler) {
        this.channelHandlerMap.put(key, channelHandler);
    }

    private static class ShutdownHook extends Thread {
        private final Service service;

        public ShutdownHook(Service service) {
            this.service = service;
        }

        @Override
        public void run() {
            service.shutdown();
        }
    }
}
